<?php
/**
 * 心跳队列后台处理脚本
 * 
 * 这个脚本应该在单独的进程中运行，定期处理Redis队列中的心跳请求
 * 可以通过以下方式启动：
 * 
 * Windows:
 *   start /B php heartbeat_processor.php
 * 
 * Linux/Mac:
 *   nohup php heartbeat_processor.php > heartbeat_processor.log 2>&1 &
 */

require_once __DIR__ . '/../../vendor/autoload.php';

error_reporting(E_ALL);
ini_set('display_errors', 'on');
set_time_limit(0);

// 读取 .env
function loadDotEnv(string $path): array {
    if (!is_file($path)) return [];
    $vars = [];
    foreach (file($path) as $line) {
        $line = trim($line);
        if ($line === '' || $line[0] === '#') continue;
        $pos = strpos($line, '=');
        if ($pos === false) continue;
        $key = trim(substr($line, 0, $pos));
        $val = trim(substr($line, $pos + 1));
        if ((strlen($val) >= 2) && (($val[0] === '"' && substr($val, -1) === '"') || ($val[0] === "'" && substr($val, -1) === "'"))) {
            $val = substr($val, 1, -1);
        }
        $vars[$key] = $val;
    }
    return $vars;
}

$env = loadDotEnv(__DIR__ . '/../../.env');

// MySQL配置
$dbConfig = [
    'host' => $env['DB_MYSQL_HOST'] ?? getenv('DB_MYSQL_HOST') ?? ($env['DB_HOST'] ?? getenv('DB_HOST') ?: '127.0.0.1'),
    'port' => $env['DB_MYSQL_PORT'] ?? getenv('DB_MYSQL_PORT') ?? ($env['DB_PORT'] ?? getenv('DB_PORT') ?: '3306'),
    'database' => $env['DB_MYSQL_DATABASE'] ?? getenv('DB_MYSQL_DATABASE') ?? ($env['DB_DATABASE'] ?? getenv('DB_DATABASE') ?: 'qiandao'),
    'username' => $env['DB_MYSQL_USERNAME'] ?? getenv('DB_MYSQL_USERNAME') ?? ($env['DB_USERNAME'] ?? getenv('DB_USERNAME') ?: 'root'),
    'password' => $env['DB_MYSQL_PASSWORD'] ?? getenv('DB_MYSQL_PASSWORD') ?? ($env['DB_PASSWORD'] ?? getenv('DB_PASSWORD') ?: ''),
    'charset' => $env['DB_MYSQL_CHARSET'] ?? getenv('DB_MYSQL_CHARSET') ?? ($env['DB_CHARSET'] ?? getenv('DB_CHARSET') ?: 'utf8mb4')
];

// Redis配置
$redisConfig = [
    'host' => $env['REDIS_HOST'] ?? getenv('REDIS_HOST') ?: '127.0.0.1',
    'port' => $env['REDIS_PORT'] ?? getenv('REDIS_PORT') ?: 6379,
    'password' => $env['REDIS_PASSWORD'] ?? getenv('REDIS_PASSWORD') ?: '',
    'database' => $env['REDIS_DATABASE'] ?? getenv('REDIS_DATABASE') ?: 0
];

function logMsg(string $msg): void {
    $timestamp = date('Y-m-d H:i:s');
    echo "[{$timestamp}] {$msg}\n";
    
    $logDir = __DIR__ . '/logs';
    if (!is_dir($logDir)) @mkdir($logDir, 0777, true);
    
    $file = $logDir . '/heartbeat_processor_' . date('Y-m-d') . '.log';
    $line = "[{$timestamp}] {$msg}\n";
    file_put_contents($file, $line, FILE_APPEND);
}

function createMySQLConnection(): ?PDO {
    global $dbConfig;
    try {
        $dsn = "mysql:host={$dbConfig['host']};port={$dbConfig['port']};dbname={$dbConfig['database']};charset={$dbConfig['charset']}";
        $pdo = new PDO($dsn, $dbConfig['username'], $dbConfig['password']);
        $pdo->setAttribute(PDO::ATTR_ERRMODE, PDO::ERRMODE_EXCEPTION);
        $pdo->setAttribute(PDO::ATTR_DEFAULT_FETCH_MODE, PDO::FETCH_ASSOC);
        logMsg("MySQL connected: {$dbConfig['host']}:{$dbConfig['port']}/{$dbConfig['database']}");
        return $pdo;
    } catch (\Throwable $e) {
        logMsg("MySQL connection failed: " . $e->getMessage());
        return null;
    }
}

function createRedisConnection(): ?object {
    global $redisConfig;
    try {
        $redis = new Redis();
        $redis->connect($redisConfig['host'], $redisConfig['port'], 2);
        if (!empty($redisConfig['password'])) {
            $redis->auth($redisConfig['password']);
        }
        $redis->select($redisConfig['database']);
        logMsg("Redis connected: {$redisConfig['host']}:{$redisConfig['port']}/{$redisConfig['database']}");
        return $redis;
    } catch (\Throwable $e) {
        logMsg("Redis connection failed: " . $e->getMessage());
        return null;
    }
}

function updateDeviceHeartbeat(PDO $pdo, string $sn, ?string $ipAddr, ?string $appVersion): void {
    $now = date('Y-m-d H:i:s');
    $stmt = $pdo->prepare('UPDATE checkin_device SET last_heartbeat_time = ?, online_status = 1, ip_address = COALESCE(?, ip_address), firmware_version = COALESCE(?, firmware_version), update_time = ? WHERE sn = ?');
    $stmt->execute([$now, $ipAddr, $appVersion, $now, $sn]);
}

// 主循环
logMsg("Heartbeat processor started");

$pdo = createMySQLConnection();
$redis = createRedisConnection();

if (!$pdo || !$redis) {
    logMsg("Failed to initialize connections");
    exit(1);
}

$processedCount = 0;
$errorCount = 0;

while (true) {
    try {
        $queueKey = "heartbeat:queue";
        
        // 批量处理队列中的心跳请求（每次最多处理50个）
        $batchSize = 0;
        for ($i = 0; $i < 50; $i++) {
            $data = $redis->rPop($queueKey);
            if (!$data) break;
            
            $heartbeatData = json_decode($data, true);
            if (!is_array($heartbeatData)) continue;
            
            try {
                $sn = strval($heartbeatData['sn'] ?? '');
                $ipAddr = isset($heartbeatData['ipAddr']) ? strval($heartbeatData['ipAddr']) : null;
                $appVersion = isset($heartbeatData['appVersion']) ? strval($heartbeatData['appVersion']) : null;
                
                if ($sn !== '') {
                    updateDeviceHeartbeat($pdo, $sn, $ipAddr, $appVersion);
                    $processedCount++;
                    $batchSize++;
                }
            } catch (\Throwable $e) {
                logMsg("Error processing heartbeat: " . $e->getMessage());
                $errorCount++;
            }
        }
        
        if ($batchSize > 0) {
            logMsg("Processed {$batchSize} heartbeats (total: {$processedCount}, errors: {$errorCount})");
        }
        
        // 每秒检查一次队列
        sleep(1);
        
    } catch (\Throwable $e) {
        logMsg("Fatal error: " . $e->getMessage());
        sleep(5);
    }
}

